﻿using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Web;

namespace RabbitMQ_Test
{
    public class RabbitMQUtility
    {
        public class QMessage
        {
            public int Title { get; set; }
            public string value { get; set; }
            public DateTime dateTime { get; set; }
            public QMessage() { }
            public QMessage(DateTime dateTime, string value)
            {
                this.value = value;
                this.dateTime = dateTime;
            }
        }
        public async static Task<bool?> Producer(string value)
        {

            return await Task.Run(() =>
            {
                bool? returnValue = null;
                try
                {
                    var qName = "synchronization_table";
                    var exchangeName = "ycgwl_crm";
                    var exchangeType = "fanout";//topic、fanout
                    var routingKey = "*";
                    var uri = new Uri("amqp://172.16.251.187:5673/");
                    byte deliveryMode = 2;//硬盘序列化
                    ushort heartbeatTime = 60;
                    ConnectionFactory factory = new ConnectionFactory
                    {
                        HostName = "rabbitmq1.rabbit.com",
                        UserName = "ycgwlrmq",
                        Password = "123abc~~~",
                        RequestedHeartbeat = heartbeatTime,
                        Endpoint = new AmqpTcpEndpoint(uri),
                        VirtualHost = "ycgwl_crm_vhost",
                        AutomaticRecoveryEnabled = true
                    };
                    using (var connection = factory.CreateConnection())
                    {
                        using (var channel = connection.CreateModel())
                        {
                            //设置交换器的类型
                            channel.ExchangeDeclare(exchangeName, exchangeType);
                            //声明一个队列，设置队列是否持久化，排他性，与自动删除
                            channel.QueueDeclare(qName, true, false, false, null);
                            //绑定消息队列，交换器，routingkey
                            channel.QueueBind(qName, exchangeName, routingKey);

                            //队列持久化
                            IBasicProperties properties = channel.CreateBasicProperties();
                            properties.Persistent = true;
                            properties.DeliveryMode = deliveryMode;//硬盘序列化
                            var m = new QMessage(DateTime.Now, value);
                            var body = Encoding.UTF8.GetBytes(Newtonsoft.Json.JsonConvert.SerializeObject(m));
                            //发送信息
                            try
                            {
                                channel.BasicPublish(exchangeName, routingKey, properties, body);
                                returnValue = true;
                            }
                            catch (Exception oe)
                            {
                                string err = oe.Message + oe.StackTrace;
                            }

                        }
                    }
                }
                catch (Exception oe)
                {
                    string err = oe.Message + oe.StackTrace;
                }
                return returnValue;
            });
        }

        public async static Task<string> Consumer()
        {
            return await Task.Run(() =>
            {
                string value = null;
                try
                {
                    var qName = "synchronization_table_return";
                    var exchangeName = "ycgwl_crm";
                    var exchangeType = "fanout";//topic、fanout
                    var routingKey = "*";
                    var uri = new Uri("amqp://172.16.251.187:5673/");
                    ushort heartbeatTime = 60;
                    var factory = new ConnectionFactory
                    {
                        HostName = "rabbitmq1.rabbit.com",
                        UserName = "ycgwlrmq",
                        Password = "123abc~~~",
                        RequestedHeartbeat = heartbeatTime,
                        Endpoint = new AmqpTcpEndpoint(uri),
                        VirtualHost = "ycgwl_crm_vhost",
                        AutomaticRecoveryEnabled = true
                    };
                    using (var connection = factory.CreateConnection())
                    {
                        using (var channel = connection.CreateModel())
                        {
                            channel.ExchangeDeclare(exchangeName, exchangeType);
                            channel.QueueDeclare(qName, true, false, false, null);
                            channel.QueueBind(qName, exchangeName, routingKey);
                            //定义这个队列的消费者
                            QueueingBasicConsumer consumer = new QueueingBasicConsumer(channel);
                            //false为手动应答，true为自动应答
                            channel.BasicQos(0, 1, false);
                            channel.BasicConsume(qName, false, consumer);
                            while (true)
                            {
                                try
                                {
                                    BasicDeliverEventArgs ea = null;
                                    QMessage message = null;
                                    string messageStr = null;
                                    try
                                    {
                                        ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
                                        byte[] bytes = ea.Body;
                                        messageStr = Encoding.UTF8.GetString(bytes);
                                        message = Newtonsoft.Json.JsonConvert.DeserializeObject<QMessage>(messageStr);
                                        value = messageStr;
                                    }
                                    catch (Exception oe)
                                    {
                                        string err = oe.Message + oe.StackTrace;
                                    }
                                    Console.WriteLine("Receive a Message, DateTime:" + message.dateTime.ToString("yyyy-MM-dd HH:mm:ss") + " Title:" + message.Title);
                                    try
                                    {
                                        //channel.BasicAck(ea.DeliveryTag, false);//这条应该是删除消息的语句 配合插入同内容流水id的结果队列的内容 应该就可以完成一个闭合  暂时没找到标记已经消费的。。。这个到时候要测试一下
                                        ////如果是自动应答，下下面这句代码不用写啦。
                                        //if ((Convert.ToInt32(message.Title) % 2) == 1)
                                        //{
                                        //channel.BasicAck(ea.DeliveryTag, false);
                                        //}
                                    }
                                    catch (Exception oe)
                                    {
                                        string err = oe.Message + oe.StackTrace;
                                    }
                                }
                                catch (Exception oe)
                                {
                                    string err = oe.Message + oe.StackTrace;
                                }
                                break;
                            }
                        }
                    }
                }
                catch (Exception oe)
                {
                    string err = oe.Message + oe.StackTrace;
                }
                return value;
            });
        }
    }
}